package com.dobbinsoft.iotechn.device.mqtt;

import com.dobbinsoft.iotechn.data.logger.Logger;
import com.dobbinsoft.iotechn.data.logger.LoggerFactory;
import com.dobbinsoft.iotechn.device.constant.DeviceConst;
import com.dobbinsoft.iotechn.device.enums.IotMessageResultType;
import com.dobbinsoft.iotechn.device.enums.IotMessageType;
import com.dobbinsoft.iotechn.device.mqtt.handler.IotMqttMessageHandler;
import com.dobbinsoft.iotechn.device.mqtt.message.IotSuperMessage;
import com.dobbinsoft.iotechn.device.mqtt.serializer.IotMqttDeserializer;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.Map;

/**
 * Description:
 * User: rize
 * Date: 2020/6/22
 * Time: 14:12
 */
public class IotMqttCallback implements MqttCallbackExtended {

    private static final Logger logger = LoggerFactory.getLogger(IotMqttCallback.class);

    private String topic;

    private MqttClient mqttClient;

    // 不同的消息有不同的处理器，通过消息类型来映射
    private Map<Integer, IotMqttMessageHandler> messageHandleRouter;

    // 所有消息均是使用同一套序列化/反序列化协议
    private IotMqttDeserializer iotMqttDeserializer;

    private IotMqttManagerClient iotMqttManagerClient;

    public IotMqttCallback(MqttClient mqttClient, Map<Integer, IotMqttMessageHandler> messageHandleRouter, IotMqttDeserializer iotMqttDeserializer, String topic, IotMqttManagerClient iotMqttManagerClient) {
        this.mqttClient = mqttClient;
        this.messageHandleRouter = messageHandleRouter;
        this.iotMqttDeserializer = iotMqttDeserializer;
        this.topic = topic;
        this.iotMqttManagerClient = iotMqttManagerClient;
    }

    @Override
    public void connectComplete(boolean b, String s) {
        logger.info("[MQTT 应用服务器] 尝试连接MQTT成功！");
    }

    @Override
    public void connectionLost(Throwable throwable) {
        logger.error("[MQTT 应用服务器] 失去连接" , throwable);
        try {
            mqttClient.connect();
            mqttClient.subscribe(this.topic);
        } catch (Exception e) {
            logger.error("[MQTT 应用服务器] 尝试重新连接异常", e);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        // 对消息进行反序列化操作。此操作是抽象操作，应该抽个接口。
        try {
            String accessKeyId = topic.substring(DeviceConst.TOPIC_UP.length());
            IotSuperMessage superMessage = iotMqttDeserializer.parseObject(mqttMessage.getPayload(), IotSuperMessage.class);
            if (superMessage.getCode() == IotMessageType.TO_DEVICE.getCode()) {
                return;
            }
            logger.info("[MQTT 应用服务器] 接收到消息: AccessKeyId:" + accessKeyId + ";message:" + new String(mqttMessage.getPayload(), "utf-8"));
            IotMqttMessageHandler iotMqttMessageHandler = messageHandleRouter.get(superMessage.getCode());
            // 这里反序列化两次，但是不反序列化两次，又没办法获取里面的Code。
            Object object = iotMqttDeserializer.parseObject(mqttMessage.getPayload(), iotMqttMessageHandler.getMessageType());
            int resCode = iotMqttMessageHandler.handle(accessKeyId, object);
            if (resCode == IotMessageResultType.ERROR_FATAL.getCode()) {
                // 作为客户端是没有权限将设备踢下线的，只能通过SDK或HTTP接口，通知MQ Broker将设备踢下线
                // 由于Broker是抽象的，所以这里需要抽取一个接口来
                iotMqttManagerClient.deleteDevice(accessKeyId);
                // 这里在合适的情况下可以新增禁用功能
                // iotMqttManagerClient.banDevice(topic, 5);
                logger.info("[MQTT 应用服务器] 设备被强迫下线 AccessKeyId=" + accessKeyId);
            }
        } catch (Exception e) {
            logger.error("[MQTT 应用服务器] 处理消息 异常 topic=" + topic + ";message", e);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        logger.info("[MQTT 应用服务器] 发送消息成功");
    }

}
